package spark.stream.mr;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

public final class JavaMRStream {
    private JavaMRStream() {
    }

    public static void main(String[] args) throws Exception {
        SparkConf sparkConf = new SparkConf().setAppName("MRStream");
        JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, new Duration(1000));
        final String dataDirectory = "hdfs://Master:9000/library/SparkStreaming/Data";
        JavaDStream lines = ssc.textFileStream(dataDirectory);
        JavaDStream<MrBean> mrStream = lines.map(new Function<String, MrBean>() {
            public MrBean call(String s) throws Exception {
                return new MrBean(s);
            }
        });

        JavaPairDStream<String, MrBean> mrPairStream = mrStream.mapToPair(new PairFunction<MrBean, String, MrBean>() {
            public Tuple2<String, MrBean> call(MrBean mrBean) throws Exception {
                return new Tuple2<String, MrBean>(mrBean.getCellId(), mrBean);
            }
        });


        JavaPairDStream<String, MrBean> reducedStream = mrPairStream.reduceByKey(
                new Function2<MrBean, MrBean, MrBean>() {
                    public MrBean call(MrBean mr1, MrBean mr2) throws Exception {
                        return mr1.reduce(mr2);
                    }
                }
        );
        ssc.start();
        ssc.awaitTermination();
    }
}
